package beautiful.butterfly.distributed_task_processing_engine.task_manage;

import beautiful.butterfly.distributed_task_processing_engine.jvm_instance_info_and_task_processor_allocation_strategy.JvmInstanceIdAndStatus;
import beautiful.butterfly.distributed_task_processing_engine.jvm_instance_info_and_task_processor_allocation_strategy.JvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager;
import beautiful.butterfly.distributed_task_processing_engine.jvm_instance_info_and_task_processor_allocation_strategy.TaskId$JvmInstance$ProcessorNum;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.task_data_item_data_manager.ITaskAndTaskProcessorAndTaskDataItemManage;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.task_data_item_data_manager.TaskAndTaskProcessorAndTaskDataItemManage;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor.ITaskProcessor;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor.TaskProcessor;
import beautiful.butterfly.distributed_task_processing_engine.tools.Allots;
import beautiful.butterfly.distributed_task_processing_engine.tools.Constants;
import beautiful.butterfly.distributed_task_processing_engine.tools.LocalHostAddress;
import beautiful.butterfly.distributed_task_processing_engine.tools.SpringApplicationContexttUtils;
import beautiful.butterfly.distributed_task_processing_engine.user_definition.TaskProcessorAllocationStrategyUserDefinition;
import beautiful.butterfly.distributed_task_processing_engine.zookeeper.ZooKeeperManager;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.context.ApplicationContext;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 调度服务器构造器
 */
@Slf4j
public class JvmInstanceStatusAndTaskManage
{
    //基本信息
    @Setter
    @Getter
    private String ip;
    @Setter
    @Getter
    private String hostName;
    @Setter
    @Getter
    private String jvmInstanceId;//一个实例绑定一个TaskScheduleFactory
    //zooKeeperManager
    @Setter
    @Getter
    private Map<String, String> zooKeeperConfigMap;
    @Setter
    @Getter
    public ZooKeeperManager zooKeeperManager;


    //重启处理器相关对象
    /**
     * 是否启动调度管理，如果只是做系统管理，应该设置为false
     */
    @Setter
    @Getter
    public boolean start = true;
    protected Lock lock = new ReentrantLock();
    @Setter
    @Getter
    volatile String errorMessage = "No config Zookeeper connect infomation";
    //核心处理对象
    @Setter
    @Getter
    private JvmInstanceIdStatusAndTaskManageInitThread jvmInstanceIdStatusAndTaskManageInitThread;
    @Setter
    @Getter
    private Timer timer;
    @Setter
    @Getter
    private int timerInterval = 2000;
    @Setter
    @Getter
    private JvmInstanceIdStatusAndTaskManageTimerTask jvmInstanceIdStatusAndTaskManageTimerTask;
    /**
     * ManagerFactoryTimerTask上次执行的时间戳。<br/>
     * zk环境不稳定，可能导致所有task自循环丢失，调度停止。<br/>
     * 外层应用，通过jmx暴露心跳时间，监控这个tbschedule最重要的大循环。<br/>
     */
    @Setter
    @Getter
    public volatile long timerTaskHeartBeatTimeMillis = System.currentTimeMillis();

    //任务处理
    @Setter
    @Getter
    private ITaskAndTaskProcessorAndTaskDataItemManage taskProcessorAndTaskDataItemManage;
    @Setter
    @Getter
    private JvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager;
    @Setter
    @Getter
    private Map<String, List<ITaskProcessor>> taskIdToTaskProcessorListMap = new ConcurrentHashMap<String, List<ITaskProcessor>>();
    //

    public JvmInstanceStatusAndTaskManage()
    {
        this.ip = LocalHostAddress.getLocalHostAddress();
        this.hostName = LocalHostAddress.getLocalHostName();
    }

    //初始入口
    public void init() throws Exception
    {
        Properties properties = new Properties();
        for (Map.Entry<String, String> entry : this.zooKeeperConfigMap.entrySet())
        {
            properties.put(entry.getKey(), entry.getValue());
        }
        this.init(properties);
    }

    public void init(Properties properties) throws Exception
    {
        if (Constants.debug)
        {
            log.warn("配置信息:" + properties);
        }
        if (this.jvmInstanceIdStatusAndTaskManageInitThread != null)
        {
            this.jvmInstanceIdStatusAndTaskManageInitThread.stopThread();
        }
        //
        this.lock.lock();
        try
        {
            this.taskProcessorAndTaskDataItemManage = null;
            this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager = null;
            if (this.zooKeeperManager != null)
            {
                this.zooKeeperManager.close();
            }
            //
            DistributedTaskProcessingEngine.jvmInstanceStatusAndTaskManage = this;
            this.zooKeeperManager = new ZooKeeperManager(properties);
            jvmInstanceIdStatusAndTaskManageInitThread = new JvmInstanceIdStatusAndTaskManageInitThread(this);
            jvmInstanceIdStatusAndTaskManageInitThread.setName(JvmInstanceIdStatusAndTaskManageInitThread.class.getSimpleName());
            jvmInstanceIdStatusAndTaskManageInitThread.start();
        } finally
        {
            this.lock.unlock();
        }
    }

    /**
     * 在Zk状态正常后进行数据初始化
     */
    public void doInit() throws Exception
    {
        log.warn(this.getClass().getName() + " 在Zk状态正常后进行数据初始化");
        this.zooKeeperManager.checkParentPathHasScheduleTagAndCheckTaskEngineVersion();
        //

        this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager = new JvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager(this.zooKeeperManager);
        this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager.registerJvmInstanceInfoToZooKeeperAndReturnDeletedTaskIdList(this);
        //
        this.taskProcessorAndTaskDataItemManage = new TaskAndTaskProcessorAndTaskDataItemManage(this.zooKeeperManager);
        //
        if (this.timer == null)
        {
            this.timer = new Timer();
        }
        if (this.jvmInstanceIdStatusAndTaskManageTimerTask == null)
        {
            this.jvmInstanceIdStatusAndTaskManageTimerTask = new JvmInstanceIdStatusAndTaskManageTimerTask(this);
            this.timer.schedule(this.jvmInstanceIdStatusAndTaskManageTimerTask, 2000, this.timerInterval);
        }
        this.start = true;//init ok
    }

    public void reInit(Properties properties) throws Exception
    {
        if (this.start == true || this.timer != null || this.taskIdToTaskProcessorListMap.size() > 0)
        {
            throw new Exception("");
        }
        this.init(properties);
    }


    public void jvmInstanceStatusAndTaskManage() throws Exception
    {
        log.warn("进行状态管理,如果出现异常则清除任务重新建立任务,如果是jvm暂停则只需要清除任务");
        this.lock.lock();
        try
        {
            JvmInstanceIdAndStatus jvmInstanceIdAndStatus = null;
            boolean isException = false;
            try
            {
                jvmInstanceIdAndStatus = this.getJvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager().getJvmInstanceIdAndStartStatus(this.getJvmInstanceId());
            } catch (Exception e)
            {
                isException = true;
                log.warn("异常:" + e.getMessage());
                log.error(e.getMessage());
            }
            //
            if (isException)//异常
            {
                log.error("当前" + this.getClass().getSimpleName() + "获取机器信息失败,清理处理器分配策略任务,撤销当前" + this.getClass().getSimpleName() + "在处理器分配策略和机器信息管理器中的信息注册");
                try
                {
                    stopTaskProcessorList(null);
                    this.getJvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager().deleteTaskId$JvmInstanceTaskInfosWithJvmInstanceId(this);
                } finally
                {
                    rebuildTasks();//rebuildTasks
                }
            } else if (!jvmInstanceIdAndStatus.isActive())//暂停
            {
                stopTaskProcessorList(null);
                this.getJvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager().deleteTaskId$JvmInstanceTaskInfosWithJvmInstanceId(this);
                //不用重新构建任务
            } else
            {
                //正常的操作
                rebuildTasks();//rebuildTasks
            }
        } finally
        {
            this.lock.unlock();
        }
    }


    public void rebuildTasks() throws Exception
    {
        //如果被暂停或者分配错误则需要删除
        List<String> deletedTaskIdList = this.getJvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager().registerJvmInstanceInfoToZooKeeperAndReturnDeletedTaskIdList(this);
        for (String deletedTaskId : deletedTaskIdList)
        {
            this.stopTaskProcessorList(deletedTaskId);//清理没有暂停的任务
        }
        //stop first,and then rebuild
        this.updateTaskId$JvmInstanceTaskInfoRequestableNum();
        this.rebuildTaskProcessorsAsAllotsPlan();
    }

    public void stopTaskProcessorList(String taskId) throws Exception
    {
        //all
        if (taskId == null)
        {
            log.warn("停止所有任务的处理器");
            String[] taskIdList = this.taskIdToTaskProcessorListMap.keySet().toArray(new String[0]);
            for (String $taskId : taskIdList)
            {
                for (ITaskProcessor taskProcessor : this.taskIdToTaskProcessorListMap.get($taskId))
                {
                    try
                    {
                        taskProcessor.stop();
                    } catch (Throwable e)
                    {
                        log.error("taskId:" + taskId, e);
                    }
                }
                this.taskIdToTaskProcessorListMap.remove($taskId);
            }
        } else
        {//one
            log.warn("停止任务:" + taskId + "的处理器");
            List<ITaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.get(taskId);
            if (taskProcessorList != null)
            {
                for (ITaskProcessor taskProcessor : taskProcessorList)
                {
                    try
                    {
                        taskProcessor.stop();
                    } catch (Throwable e)
                    {
                        log.error("taskId:" + taskId, e);
                    }
                }
                this.taskIdToTaskProcessorListMap.remove(taskId);
            }

        }
    }

    /**
     * 控制中心进行拆分任务
     */
    public void updateTaskId$JvmInstanceTaskInfoRequestableNum() throws Exception
    {
        for (TaskId$JvmInstance$ProcessorNum taskId$JvmInstanceInfo : //
                this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager.getTaskId$JvmInstanceTaskInfoListByJvmInstanceId(this.jvmInstanceId))
        {
            List<TaskId$JvmInstance$ProcessorNum> taskId$JvmInstanceInfoList = this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager.getTaskId$JvmInstance$ProcessorNumList(taskId$JvmInstanceInfo.getTaskId());
            if (taskId$JvmInstanceInfoList.size() == 0 || !this.isLeader(this.jvmInstanceId, taskId$JvmInstanceInfoList))
            {
                continue;
            }
            //TaskId中心服务器进行重新分配

            TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition = this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager.getTaskProcessorAllocationStrategyUserDefinition(taskId$JvmInstanceInfo.getTaskId());
            log.warn("加载处理器分配策略:" + taskId$JvmInstanceInfo.getTaskId() + ",在jvm上进行设置任务并行执行的数量");
            //int taskItemNum, int serverNum
            int[] taskNumbers = Allots.allot(taskProcessorAllocationStrategyUserDefinition.getMaxAllowThreadGroupNum(), taskId$JvmInstanceInfoList.size());//serverNum
            for (int i = 0; i < taskId$JvmInstanceInfoList.size(); i++)
            {  //更新请求的服务器数量
                //更新JvmInstanceId可以执行TaskId任务的并行数
                this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager.updateTaskId$JvmInstance$ProcessorNum(//
                        taskId$JvmInstanceInfoList.get(i).getTaskId(),//
                        taskId$JvmInstanceInfoList.get(i).getJvmInstanceId(),//
                        taskNumbers[i]);//每个机器上面分配的数量
            }
        }
    }

    public boolean isLeader(String jvmInstanceId, List<TaskId$JvmInstance$ProcessorNum> taskId$JvmInstanceInfoList)
    {
        long num = Long.parseLong(jvmInstanceId.substring(jvmInstanceId.lastIndexOf("$") + 1));
        for (TaskId$JvmInstance$ProcessorNum taskId$JvmInstanceInfo : taskId$JvmInstanceInfoList)
        {
            if (num > Long.parseLong(taskId$JvmInstanceInfo.getJvmInstanceId().substring(taskId$JvmInstanceInfo.getJvmInstanceId().lastIndexOf("$") + 1)))
            {
                return false;
            }
        }
        //<= true
        return true;
    }

    /**
     * 根据处理器分配策略计划进行重新生成可运行的任务
     */
    public void rebuildTaskProcessorsAsAllotsPlan() throws Exception
    {
        for (TaskId$JvmInstance$ProcessorNum taskId$JvmInstanceInfo : this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager.getTaskId$JvmInstanceTaskInfoListByJvmInstanceId(this.jvmInstanceId))
        {
            String taskId = taskId$JvmInstanceInfo.getTaskId();
            List<ITaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.get(taskId);//同时处理的数量
            if (taskProcessorList == null)
            {
                taskProcessorList = new ArrayList<ITaskProcessor>();
                this.taskIdToTaskProcessorListMap.put(taskId, taskProcessorList);
            }
            log.warn("加载处理器分配策略:" + taskId + "进行任务重新分发");
            int requestableNum = taskId$JvmInstanceInfo.getProcessorNum();//该task允许最大可以创建并行执行的任务
            //先减
            while (taskProcessorList.size() > 0 && taskProcessorList.size() > requestableNum)
            {
                ITaskProcessor taskProcessor = taskProcessorList.remove(taskProcessorList.size() - 1);
                try
                {
                    taskProcessor.stop();
                } catch (Throwable e)
                {
                    log.error("注销处理器失败,taskId:" + taskId, e);
                }
            }
            //后加
            TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition = this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager.getTaskProcessorAllocationStrategyUserDefinition(taskId);
            while (taskProcessorList.size() < requestableNum)
            {
                ITaskProcessor taskProcessor = this.createTaskProcessor(taskProcessorAllocationStrategyUserDefinition);//new object and init


                if (null == taskProcessor)
                {
                    log.error("创建处理器失败,taskId:" + taskId);
                } else
                {
                    taskProcessorList.add(taskProcessor);
                }
            }
        }
    }

    /**
     * 创建调度服务器
     */
    public ITaskProcessor createTaskProcessor(TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition)
            throws Exception
    {

        ITaskProcessor taskProcessor = null;
        try
        {
            taskProcessor = new TaskProcessor(this, taskProcessorAllocationStrategyUserDefinition.getTaskId(), taskProcessorAndTaskDataItemManage);
        } catch (Exception e)
        {
            e.printStackTrace();
            log.error(e.getMessage());
        }
        return taskProcessor;
    }

    public void stopAll() throws Exception
    {
        try
        {
            lock.lock();
            this.start = false;
            if (this.jvmInstanceIdStatusAndTaskManageInitThread != null)
            {
                this.jvmInstanceIdStatusAndTaskManageInitThread.stopThread();
            }
            if (this.timer != null)
            {
                if (this.jvmInstanceIdStatusAndTaskManageTimerTask != null)
                {
                    this.jvmInstanceIdStatusAndTaskManageTimerTask.cancel();
                    this.jvmInstanceIdStatusAndTaskManageTimerTask = null;
                }
                this.timer.cancel();
                this.timer = null;
            }
            this.stopTaskProcessorList(null);
            if (this.zooKeeperManager != null)
            {
                this.zooKeeperManager.close();
            }
            if (this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager != null)
            {
                try
                {
                    ZooKeeper zooKeeper = this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager.getZooKeeper();
                    if (zooKeeper != null)
                    {
                        zooKeeper.close();
                    }
                } catch (Exception e)
                {
                    log.error("stopAll zookeeper getZooKeeper", e);
                }
            }
            this.jvmInstanceId = null;
            log.warn("stopAll");
        } catch (Throwable e)
        {
            log.error("stopAll" + e.getMessage(), e);
        } finally
        {
            lock.unlock();
        }
    }


    public void reStart() throws Exception
    {
        log.warn("重新初始化任务管理:" + this.getClass().getSimpleName() + "首先进行资源清理,然后再进行资源初始化");
        try
        {
            if (this.timer != null)
            {
                if (this.jvmInstanceIdStatusAndTaskManageTimerTask != null)
                {
                    this.jvmInstanceIdStatusAndTaskManageTimerTask.cancel();
                    this.jvmInstanceIdStatusAndTaskManageTimerTask = null;//对象并行处理模型
                }
                this.timer.purge();
                this.timer = null;
            }
            //
            this.stopTaskProcessorList(null);
            //
            if (this.zooKeeperManager != null)
            {
                this.zooKeeperManager.close();
            }
            this.jvmInstanceId = null;
            //
            this.init();
        } catch (Throwable e)
        {
            log.error("" + e.getMessage(), e);
        }
    }


    public ITaskAndTaskProcessorAndTaskDataItemManage getTaskAndTaskProcessorAndTaskDataItemManage()
    {
        if (this.taskProcessorAndTaskDataItemManage == null)
        {
            throw new RuntimeException(this.errorMessage);
        }
        return taskProcessorAndTaskDataItemManage;
    }

    public JvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager getJvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager()
    {
        if (this.jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager == null)
        {
            throw new RuntimeException(this.errorMessage);
        }
        return jvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager;
    }


    public Object getBean(String beanName)
    {
        //这样可以跨越spring和原生java模型
        ApplicationContext applicationContext = SpringApplicationContexttUtils.getApplicationContext();
        return applicationContext.getBean(beanName);
    }
}

